package com.idanchuang.count.listener;

import com.alibaba.fastjson.JSON;
import com.idanchuang.count.core.constant.UserCountEventTypeEnum;
import com.idanchuang.count.service.MsgUnreadCountService;
import com.idanchuang.count.service.UserCountService;
import com.idanchuang.count.support.UserCountChecker;
import com.idanchuang.count.support.MetricCollector;
import com.idanchuang.user.count.service.api.request.command.BaseUpdateCommand;
import com.idanchuang.user.count.service.api.request.command.count.UserMultiScenesCountUpdateCommand;
import com.idanchuang.user.count.service.api.request.command.count.UserSceneCountUpdateCommand;
import com.idanchuang.user.count.service.api.request.command.read.AddUserMsgCommand;
import com.idanchuang.user.count.service.api.request.command.read.BatchAddUserMsgCommand;
import com.idanchuang.user.count.service.api.request.command.read.UserMsgAllReadCommand;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Collections;
import java.util.List;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import static com.idanchuang.count.core.util.JacksonUtil.parseObject;

/**
 * @author Richard_yyf
 * @version 1.0 2021/5/28
 */
@Component
@Slf4j
public class UserCountEventListener {

    @Resource
    private UserCountChecker userCountChecker;

    @Resource
    private UserCountService userCountService;

    @Resource
    private MsgUnreadCountService msgUnreadCountService;

    private <T extends BaseUpdateCommand> void commonConsume(Consumer<List<T>> commandConsumer,
                                                             Class<T> targetClazz,
                                                             List<ConsumerRecord<String, String>> records,
                                                             UserCountEventTypeEnum countEventType,
                                                             Acknowledgment ack) {
        long start = System.currentTimeMillis();

        List<T> commands = records
                .stream()
                // 反序列化
                .map(r -> parseObject(r.value(), targetClazz))
                // 做幂等过滤
                .filter(this::eventNotConsumed)
//                .filter(this::timeFilter)
                .collect(Collectors.toList());

        // TODO 临时log 后期删掉
        log.info("消费事件类型: {}, size: {}, 具体内容: {}", countEventType.getTag(), records.size(), JSON.toJSONString(commands));

        try {
            commandConsumer.accept(commands);
        } catch (Exception e) {
            String errMsg = String.format("消息消费处理失败, msg content:[%s]", records.get(0));
            log.error(errMsg, e);
            throw e;
        }

        ack.acknowledge();
        MetricCollector.collectAfterEventConsumed(countEventType.getTag(), commands.size(), start);
    }

    @KafkaListener(
            topics = "#{'${count.topic-prefix}' + '_user_scene_count_update'}",
            groupId = "${spring.kafka.consumer.group-id}",
            containerFactory = "customContainerFactory"
    )
    public void consumeUserSceneCountUpdate(List<ConsumerRecord<String, String>> records,
                                            Acknowledgment ack) {

        commonConsume(
                commands -> userCountService.count(commands),
                UserSceneCountUpdateCommand.class,
                records,
                UserCountEventTypeEnum.UserSceneCountUpdate,
                ack
        );
    }

    @KafkaListener(
            topics = "#{'${count.topic-prefix}' + '_user_multi_scenes_count_update'}",
            groupId = "${spring.kafka.consumer.group-id}",
            containerFactory = "customContainerFactory"
    )
    public void consumeUserMultiScenesCountUpdate(List<ConsumerRecord<String, String>> records,
                                                  Acknowledgment ack) {

        commonConsume(
                commands -> userCountService.multiCount(commands),
                UserMultiScenesCountUpdateCommand.class,
                records,
                UserCountEventTypeEnum.UserMultiScenesCountUpdate,
                ack
        );

    }

    @KafkaListener(
            topics = "#{'${count.topic-prefix}' + '_add_user_msg'}",
            groupId = "#{'${spring.kafka.consumer.group-id}'}",
            containerFactory = "customContainerFactory"
    )
    public void consumeAddUserMsg(List<ConsumerRecord<String, String>> records,
                                  Acknowledgment ack) {

        commonConsume(
                commands -> msgUnreadCountService.addUserMsgInAsyncBatch(commands),
                AddUserMsgCommand.class,
                records,
                UserCountEventTypeEnum.AddUserMsg,
                ack
        );

    }

    @KafkaListener(
            topics = "#{'${count.topic-prefix}' + '_batch_add_user_msg'}",
            groupId = "#{'${spring.kafka.consumer.group-id}'}",
            containerFactory = "customSingleContainerFactory"
    )
    public void consumeBatchAddUserMsg(ConsumerRecord<String, String> record,
                                  Acknowledgment ack) {

        commonConsume(
                commands -> msgUnreadCountService.batchAddUserMsgInAsyncBatch(commands),
                BatchAddUserMsgCommand.class,
                Collections.singletonList(record),
                UserCountEventTypeEnum.BatchAddUserMsg,
                ack
        );

    }

    @KafkaListener(
            topics = "#{'${count.topic-prefix}' + '_user_msg_all_read'}",
            groupId = "#{'${spring.kafka.consumer.group-id}'}",
            containerFactory = "customContainerFactory"
    )
    public void consumeUserAllRead(List<ConsumerRecord<String, String>> records,
                                   Acknowledgment ack) {
        commonConsume(
                commands -> msgUnreadCountService.readAllInAsyncBatch(commands),
                UserMsgAllReadCommand.class,
                records,
                UserCountEventTypeEnum.UserMsgAllRead,
                ack
        );
    }

    private boolean eventNotConsumed(BaseUpdateCommand command) {
        if (userCountChecker.checkUidExist(command.getUid(), command.getBizCode())) {
            log.info("uid[{}] 触发幂等，skip", command.getUid());
            return false;
        }
        return true;
    }

//    private boolean timeFilter(BaseUpdateCommand command) {
//        // 只消费对应时间戳之后的消息
//        if (userCountChecker.checkTimeLegal(command.getTimeStamp(), command.getBizCode())) {
//            return true;
//        } else {
//            log.info("uid[{}] 触发 time filter，skip", command.getUid());
//            return false;
//        }
//    }

}
